﻿
#include <stdlib.h>
#include <stdio.h>
#include <memory.h>
#include <assert.h>

#define LP_CORE

#include "lp_service.h"
#include "lp_thread.h"
#include "lp_timer.h"

#define DEFAULT_QUEUE_SIZE      32
#define SERVICE_WAITMS      5000
#define MSG_DATA_BUFF      0


typedef struct Service Service;
struct Service{
    LPSpinLock lock;
    uint32_t cap;
    uint32_t head;
    uint32_t tail;
    uint32_t queueCount;
    Service * next;
    int inGlobal;
    int isWorker; 
    int exit;
    LPMsg * queue;
    LPMsg cacheQueue[DEFAULT_QUEUE_SIZE];
    LPService h;
    void * ud;
    LPSession sessionIdx;
    LPMsgCallback msgCallback;
    LPExitSvrCallback exitCallback;
};


static struct{
    Service * head;
    Service * tail;
    LPSpinLock queueLock;
    LPStorage * storage;
    LPRWLock storageLock;
}gCtx = {0};

static void
pushGlobalQueue (Service * s){
    assert(s->next == NULL);
    lpLockSpin(&gCtx.queueLock);
    if(gCtx.tail) {
        gCtx.tail->next = s;
        gCtx.tail = s;
    } else {
        gCtx.head = gCtx.tail = s;
    }
    lpUnlockSpin(&gCtx.queueLock);
}


static Service *
popGlobalQueue (){
    lpLockSpin(&gCtx.queueLock);
    Service * s = gCtx.head;
    if(s){
        gCtx.head = s->next;
        if(!gCtx.head){
            assert(s == gCtx.tail);
            gCtx.tail = NULL;
        }
        s->next = NULL;
    }
    lpUnlockSpin(&(gCtx.queueLock));
    return s;
}

static LPService
regInStorage (Service ** s){
    LPService h;
    lpLockRWLockW(&gCtx.storageLock);
    h = lpRegInStorage(gCtx.storage, (void **)s);
    lpUnlockRWLockW(&gCtx.storageLock);
    return h;
}

static Service *
findService (LPService h){
    Service * s;
    lpLockRWLockR(&gCtx.storageLock);
    s = lpFindInStorage(gCtx.storage, h);
    lpUnlockRWLockR(&gCtx.storageLock);
    return s;
}

static void
revServiceHandle (LPService h){
    lpLockRWLockW(&gCtx.storageLock);
    lpRecInStorage(gCtx.storage, h);
    lpUnlockRWLockW(&gCtx.storageLock);
}


static Service *
createService (Service* s) {
    lpInitSpin(&s->lock);
    s->cap = DEFAULT_QUEUE_SIZE;
    s->queue = malloc(sizeof(LPMsg) * s->cap);
    s->next = NULL;
    s->sessionIdx = 10000;
    s->isWorker = 0;
    s->queueCount = 0;
    return s;
}

static void
expandServiceQueue (Service * s, uint32_t cap){
    if (cap <= DEFAULT_QUEUE_SIZE) {
        return;
    }
    LPMsg* queue = malloc(sizeof(LPMsg) * cap);
    for (unsigned i = 0; i < s->cap; i++) {
        queue[i] = s->queue[(s->head + i) % s->cap];
    }

    if (s->cap > DEFAULT_QUEUE_SIZE) {
        free(s->queue);
    }
    s->head = 0;
    s->tail = s->cap;
    s->cap = cap;
    s->queue = queue;
}

static void
pushServiceQueue (Service* s, LPMsgSend* send){
    lpLockSpin(&s->lock);
    s->queueCount++;
    LPMsg * msg = &(s->queue[s->tail]);
    msg->ptype = send->ptype;
    msg->cmd = send->cmd;
    msg->session = send->session;
    msg->data = send->data;
    msg->size = send->size;
    msg->free = send->free;
    if (++s->tail >= s->cap) {
        s->tail = 0;
    }
    if (s->head == s->tail) {
        expandServiceQueue(s, s->cap * 2);
    }
    if (s->isWorker && !s->inGlobal) {
        s->inGlobal = 1;
        pushGlobalQueue(s);
    }
    lpUnlockSpin(&s->lock);
}

static int
popServiceQueue (Service* s, LPMsg* msg, int max) {
    int ret = 0;
    lpLockSpin(&s->lock);
    while (s->head != s->tail && ret < max) {
        msg[ret] = s->queue[s->head++];
        s->queueCount --;
        ret += 1;
        int head = s->head;
        int tail = s->tail;
        int cap = s->cap;
        if (head >= cap) {
            s->head = head = 0;
        }
    }
    if(s->queueCount == 0 && s->cap > (DEFAULT_QUEUE_SIZE * 2)){
        free(s->queue);
        s->cap = DEFAULT_QUEUE_SIZE;
        s->queue = s->cacheQueue;
        s->head = s->tail = 0;
    }
    lpUnlockSpin(&s->lock);
    return ret;
}


static void
freeServiceQueue (Service * s){
    if (s->exitCallback) {
        s->exitCallback(s->h, s->ud);
    }
    lpClearSpin(&s->lock);
    LPMsg * msg;
    while (s->head != s->tail) {
        msg = &(s->queue[s->head++]);
        if(msg->data && msg->free){
            free((void *)msg->data);
        }
        if (s->head >= s->cap) {
            s->head = 0;
        }
    }
    free(s->queue);
    revServiceHandle(s->h);
}

void
dispatchMsg (Service * s, LPMsg * msg){
    s->msgCallback(msg);
    if (msg->free) {
        msg->free(msg);
    }
}


static inline LPSession
incSession (Service * s){
    LPSession id = ++ s->sessionIdx;
    if(id == 0){
        s->sessionIdx ++;
        return 1;
    }
    return id;
}


LP_API int
lpInitServiceContext (){
    lpInitSpin(&gCtx.queueLock);
    lpInitRWLock(&gCtx.storageLock);
    gCtx.storage = lpNewStorage(sizeof(Service), 0xFF + 1);
    return 0;
}

LP_API int
lpExitServiceContext(){
    lpClearSpin(&gCtx.queueLock);
    lpClearRWLock(&gCtx.storageLock);
    return 0;
}

LP_API LPService
lpNewService (LPMsgCallback cb, void * ud){
    Service* s;
    LPService h = regInStorage(&s);
    createService(s);
    s->msgCallback = cb;
    s->ud = ud;
    s->h = h;
    return h;
}

LP_API int
lpDispatchWorker (){
    Service* s = popGlobalQueue();
    if (!s){
        return 0;
    }
    LPMsg msg[8];
    int ret = popServiceQueue(s, msg, 8);
    if(!ret){
        return 0;
    }
    for (int i = 0; i < ret; i ++){
        dispatchMsg(s, &(msg[i]));
    }
    if (s->exit) {
        freeServiceQueue(s);
    }else {
        lpLockSpin(&s->lock);
        if (s->tail != s->head) {
            s->inGlobal = 1;
        }else {
            s->inGlobal = 0;
        }
        if (s->inGlobal) {
            pushGlobalQueue(s);
        }
        lpUnlockSpin(&s->lock);
    }
    return 1;
}


LP_API int
lpDispatchService (LPService h){
    Service* s = findService(h);
    if(!s){ return 1; }
    LPMsg msg[8];
    int ret = popServiceQueue(s, msg, 8);
    if(!ret){
        return 0;
    }
    for (int i = 0; i < ret; i ++){
        dispatchMsg(s, &(msg[i]));
    }
    return ret;
}

LP_API int
lpAttachWorker (LPService h){
    Service* s = findService(h);
    if(!s){ return 1; }
    if(s->isWorker){ return 1; }
    lpLockSpin(&s->lock);
    s->isWorker = 1;
    if(!s->inGlobal && (s->tail != s->head)){
        s->inGlobal = 1;
        pushGlobalQueue(s);
    }
    lpUnlockSpin(&s->lock);
    return 0;
}

LP_API int
lpSend (LPService h, LPMsgSend* send){
    Service* s = findService(h);
    if(!s){ return 1; }
    pushServiceQueue(s, send);
    return 0;
}

LP_API int
lpNewSession (LPService h, LPSession * session){
    Service* s = findService(h);
    if(!s){ return 1; }
    *session = incSession(s);
    return 0;
}

